package com.yjb.utils;


import com.alibaba.fastjson.JSON;

import com.alibaba.fastjson.TypeReference;
import redis.clients.jedis.Jedis;

import java.lang.reflect.Type;
import java.util.Set;
import java.util.UUID;

public class RedisDelayingQueue {

    static class TaskItem<T>{
        public String id;
        public String msg;
    }

    // fastjson 序列化对象中存在 generic 类型时，需要使用 TypeReference
    private Type TaksType = new TypeReference<TaskItem>(){}.getType();
    private Jedis jedis;
    private String queueKey;

    public RedisDelayingQueue(Jedis jedis, String queueKey) {
        this.jedis = jedis;
        this.queueKey = queueKey;
    }

    public void delay(String msg){
        TaskItem task = new TaskItem();
        task.id = UUID.randomUUID().toString();
        task.msg = msg;
        String s = JSON.toJSONString(task);  //fastjson序列化
        jedis.zadd(queueKey,System.currentTimeMillis()+5000,s);   //塞入延时队列，5s后再试
    }

    public void loop() throws Exception {
        while (!Thread.interrupted()){
            // 支取一条
            Set<String> values = jedis.zrangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
            if (values.isEmpty()){
                Thread.sleep(500);
                continue;
            }
            String s = values.iterator().next();
            if (jedis.zrem(queueKey,s)>0){  //抢到了
                TaskItem task = JSON.parseObject(s, TaskItem.class);  // 反序列化
                this.handleMsg(task.msg);
            }
        }

    }

    public void handleMsg(Object msg) throws Exception {
        System.out.println(msg);
    }

    public static void main(String[] args) {
        Jedis jedis = new Jedis();
        RedisDelayingQueue queue = new RedisDelayingQueue(jedis, "q-demo");
        Thread producer = new Thread() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++) {
                    queue.delay("codehole" + i);
                }
            }
        };
        Thread consumer = new Thread() {
            @Override
            public void run() {
                try {
                    queue.loop();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
        producer.start();
        consumer.start();
        try {
            producer.join();
            Thread.sleep(6000);
            consumer.interrupt();
            consumer.join();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}


